package cn.ztuo.bitrade.consumer;

import cn.ztuo.bitrade.component.CoinExchangeRate;
import cn.ztuo.bitrade.constant.NettyCommand;
import cn.ztuo.bitrade.entity.ContractCoin;
import cn.ztuo.bitrade.entity.ContractOrder;
import cn.ztuo.bitrade.entity.ContractTrade;
import cn.ztuo.bitrade.entity.TradePlate;
import cn.ztuo.bitrade.handler.MongoMarketHandler;
import cn.ztuo.bitrade.handler.NettyHandler;
import cn.ztuo.bitrade.handler.WebsocketMarketHandler;
import cn.ztuo.bitrade.job.ContractPushJob;
import cn.ztuo.bitrade.processor.CoinProcessor;
import cn.ztuo.bitrade.processor.CoinProcessorFactory;
import cn.ztuo.bitrade.processor.DefaultCoinProcessor;
import cn.ztuo.bitrade.service.ContractOrderDetailService;
import cn.ztuo.bitrade.service.ContractOrderService;
import cn.ztuo.bitrade.service.MarketService;
import cn.ztuo.bitrade.util.ExceptionUtil;
import cn.ztuo.bitrade.util.MessageResult;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class ContractTradeConsumer {
    @Autowired
    private ContractOrderService contractOrderService;
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    @Autowired
    private NettyHandler nettyHandler;
    @Autowired
    private CoinProcessorFactory coinProcessorFactory;
    @Autowired
    private ContractOrderDetailService exchangeOrderDetailService;
    @Autowired
    private MongoMarketHandler mongoMarketHandler;
    @Autowired
    private WebsocketMarketHandler wsHandler;
    @Autowired
    private ContractPushJob pushJob;

    @Autowired
    private MarketService marketService;
    @Autowired
    private CoinExchangeRate exchangeRate;

    @Value("${second.referrer.award}")
    private boolean secondReferrerAward;

    private ExecutorService executor = new ThreadPoolExecutor(
            30, 30, 10, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(30),
            new ThreadPoolExecutor.DiscardOldestPolicy());

    /**
     * 订单取消成功
     *
     * @param records
     */
    @KafkaListener(topics = "contract-simple-order-cancel-success", containerFactory = "kafkaListenerContainerFactory")
    public void handleOrderCanceled(List<ConsumerRecord<String, String>> records) {
        log.info("contract-simple-order-cancel-success records ====================================== {}",records);
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            try {
                log.info("contract-simple-order-cancel-success topic={},accessKey={},value={}", record.topic(), record.key(), record.value());
                String symbol = record.key();
                ContractOrder order = JSON.parseObject(record.value(), ContractOrder.class);
                //调用服务处理
                MessageResult messageResult = contractOrderService.orderCanceled(order.getOrderId(), order.getStart(), order.getTradedAmount(), order.getTurnover());
                log.info("取消订单成功messageResult={}", messageResult);
                //推送实时成交
                messagingTemplate.convertAndSend("/topic/contract-market/order-canceled/" + symbol + "/" + order.getMemberId(), order);
                nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_CANCELED, order);
            } catch (Exception e) {
                e.printStackTrace();
                log.info("取消订单ERROR={}", e);
            }
        }
    }

    /**
     * @param records
     */
    @KafkaListener(topics = "contract-market-symbol", containerFactory = "kafkaListenerContainerFactory")
    public void onAddCoinTradeByExchangeCoin(List<ConsumerRecord<String, String>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            log.info("接收contract-market-product>>topic={},value={},size={}", record.topic(), record.value(), records.size());
            List<ContractCoin> exchangeCoins = JSON.parseArray(record.value(), ContractCoin.class);
            exchangeCoins.forEach(exchangeCoin -> {
                CoinProcessor coinProcessor = coinProcessorFactory.getProcessor(exchangeCoin.getSymbol());
                if (coinProcessor == null) {
                    CoinProcessor processor = new DefaultCoinProcessor(exchangeCoin.getSymbol(), exchangeCoin.getBaseSymbol(), marketService, exchangeCoin);
                    processor.addHandler(mongoMarketHandler);
                    processor.addHandler(wsHandler);
                    processor.addHandler(nettyHandler);
                    processor.setMarketService(marketService);
                    processor.setExchangeRate(exchangeRate);
                    processor.initializeThumb();
                    processor.initializeUsdRate(2);
                    processor.setIsHalt(false);
                    coinProcessorFactory.addProcessor(exchangeCoin.getSymbol(), processor);
                }
            });
            exchangeRate.setCoinProcessorFactory(coinProcessorFactory);
        }
    }

    /**
     * 订单完成
     *
     * @param records
     */
    @KafkaListener(topics = "contract-order-completed", containerFactory = "kafkaListenerContainerFactory")
    public void handleOrderCompleted(List<ConsumerRecord<String, String>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            log.info("contract-order-completed topic={},accessKey={},value={}", record.topic(), record.key(), record.value());
            String symbol = record.key();
            try {
                List<ContractOrder> orders = JSON.parseArray(record.value(), ContractOrder.class);
                for (ContractOrder order : orders) {
                    //委托成交完成处理
                    contractOrderService.orderCompleted(order.getOrderId(), order.getStart(), order.getTradedAmount(), order.getTurnover(), order);
                    //推送订单成交
                    messagingTemplate.convertAndSend("/topic/contract-market/order-completed/" + symbol + "/" + order.getMemberId(), order);
                    nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_COMPLETED, order);
                }
            } catch (Exception e) {
                log.info(" 订单完成ERROR={}", e);
            }
        }
    }

    /**
     * 处理成交明细
     *
     * @param records
     */
    @KafkaListener(topics = "contract-trade", containerFactory = "kafkaListenerContainerFactory")
    public void handleTrade(List<ConsumerRecord<String, String>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            log.info("contract-trade topic={},key={},value={}", record.topic(), record.key(), record.value());
            long startTick = System.currentTimeMillis();
            String symbol = record.key();
            List<ContractTrade> trades = JSON.parseArray(record.value(), ContractTrade.class);
            //处理K线行情
            CoinProcessor coinProcessor = coinProcessorFactory.getProcessor(symbol);
            if (coinProcessor != null) {
                //处理新生成的交易信息 处理k线 概况和 存储交易信息
                coinProcessor.process(trades);
            }
            if(trades.get(0).getType()==null||trades.get(0).getType()!=1){
                executor.submit(new HandleTradeThread(record));
                log.info("complete exchange process,{}ms used!", System.currentTimeMillis() - startTick);
            }else{
                pushJob.addTrades(symbol, trades);
            }
        }
    }

    public class HandleTradeThread implements Runnable {
        private ConsumerRecord<String, String> record;

        private HandleTradeThread(ConsumerRecord<String, String> record) {
            this.record = record;
        }

        @Override
        public void run() {
            try {
                List<ContractTrade> trades = JSON.parseArray(record.value(), ContractTrade.class);
                String symbol = record.key();
                for (ContractTrade trade : trades) {
                    //成交明细处理
                    //推送订单成交订阅
                    contractOrderService.processExchangeTrade(trade,secondReferrerAward);
                    ContractOrder buyOrder = contractOrderService.findOne(trade.getBuyOrderId());
                    ContractOrder sellOrder = contractOrderService.findOne(trade.getSellOrderId());
                    messagingTemplate.convertAndSend("/topic/contract-market/order-trade/" + symbol + "/" + buyOrder.getMemberId(), buyOrder);
                    messagingTemplate.convertAndSend("/topic/contract-market/order-trade/" + symbol + "/" + sellOrder.getMemberId(), sellOrder);
                    nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_TRADE, buyOrder);
                    nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_TRADE, sellOrder);
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error("撮合模块ERROR={}", ExceptionUtil.getMessage(e));
            }
        }
    }


    /**
     * 消费交易盘口信息
     *
     * @param records
     */
    @KafkaListener(topics = "contract-trade-plate", containerFactory = "kafkaListenerContainerFactory")
    public void handleTradePlate(List<ConsumerRecord<String, String>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            try {
                log.info("exchange-trade-plate topic={},accessKey={}", record.topic(), record.key());
                String symbol = record.key();
                TradePlate plate = JSON.parseObject(record.value(), TradePlate.class);
                pushJob.addPlates(symbol, plate);
            } catch (Exception e) {
                log.info("消费交易盘口信息ERROR={}", e);
            }
        }
    }


}
